package com.demo;


import com.demo.model.NewUser;
import io.lettuce.core.RedisClient;
import io.lettuce.core.RedisURI;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.sync.RedisCommands;
import io.lettuce.core.api.sync.RedisHLLCommands;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.configuration.Configuration;

public class FilterOldUserFunction extends RichFilterFunction<NewUser> {

  private RedisClient redisClient;

  private StatefulRedisConnection<String, String> connection;

  private RedisHLLCommands<String, String> redisCommands;

  private String host;

  private Integer port;

  public FilterOldUserFunction() {
  }

  @Override
  public boolean filter(NewUser value) throws Exception {
    String key = "BYPUBID_".concat(value.getDist_key().split("_")[2]);
    Long result = redisCommands.pfadd(key, value.getUser_id());
    System.out.println(value + "|||" + key + ", " + result);
    return result != 0;
  }

  public FilterOldUserFunction(String host, Integer port) {
    this.host = host;
    this.port = port;
  }

  @Override
  public void open(Configuration parameters) throws Exception {
    redisClient = RedisClient.create(RedisURI.Builder.redis(host, port).build());
    connection = redisClient.connect();
    redisCommands = connection.sync();
  }

  @Override
  public void close() throws Exception {
    connection.close();
    redisClient.shutdown();
  }
}

